package storm;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import storm.bolt.OrderDetailSaveBolt;
import storm.bolt.OrderDetailSplitBolt;
import storm.bolt.OrderSaveBolt;
import storm.bolt.OrderSplitBolt;
import storm.spout.ReadOrderDetailSpout;
import storm.spout.ReadOrderSpout;

import java.util.Map;

public class OrderTopology {
    public static void main(String[] args) throws AuthorizationException, InvalidTopologyException, AlreadyAliveException {
        //TopologyBuilder是构建拓扑的类，用于指定执行的拓扑。拓扑底层是Thrift结构，由于Thrift API非常冗长，使用TopologyBuilder可以极大地简化建立拓扑的过程。
        // 具体方法见连接：https://blog.csdn.net/sanamaz/article/details/51153868
        // 创建TopologyBuilder对象
        TopologyBuilder builder=new TopologyBuilder();
        // 添加一个id为“OrderSpout”的ReadLogSpout对象,调用了createKafkSpout方法，获取订单表数据流，由元组构成
        builder.setSpout("OrderSpout", new ReadOrderSpout().createOrderSpout());
        //对订单的数据流的每个tuple进行分词处理，获取每个元组的订单唯一编号、图书ID、图书名称、图书类型、图书数量、图书单位、出版社和订单唯一id
        builder.setBolt("OrderSplitBolt", new OrderSplitBolt()).shuffleGrouping("OrderSpout");
        //对获取的信息进行输出和保存
        builder.setBolt("OrderSaveBolt", new OrderSaveBolt()).shuffleGrouping("OrderSplitBolt");

        builder.setSpout("ReadOrderDetailSpout", new ReadOrderDetailSpout().CreateOrderDetailSpout());
        //对订单细则的数据流的每个tuple进行分词处理，获取每个元组订单细则表的信息
        builder.setBolt("OrderDetailSplitBolt", new OrderDetailSplitBolt()).shuffleGrouping("ReadOrderDetailSpout");
        //对获取的信息进行输出和保存
        builder.setBolt("OrderDetailSaveBolt", new OrderDetailSaveBolt()).shuffleGrouping("OrderDetailSplitBolt");
        //Storm集群进行参数配置,Config()对象继承自HashMap，但本身封装了一些基本的配置,ppt36页
        Config config = new Config();
        //单机运行提交storm集群,ppt37有集群提交topology
        //StormSubmitter.submitTopologyWithProgressBar("jieyang", config, builder.createTopology());
        LocalCluster local = new LocalCluster();
        local.submitTopology("OrderDetailmessages",config,builder.createTopology());

    }
}
